package io.github.junxworks.junx.stream.pipeline;

import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.dsl.BasicExecutor;
import com.lmax.disruptor.dsl.Disruptor;
import io.github.junxworks.junx.core.util.ObjectUtils;
import io.github.junxworks.junx.event.impl.DisruptorEventChannel;
import lombok.Getter;
import lombok.Setter;

import java.util.List;
import java.util.Queue;

/**
 * 管道中的基础组件
 *
 * @author 王兴
 * @date 2024/09/10
 */
@Getter
@Setter
public class DisruptorComponent extends BaseComponent {

    /**
     * @return
     */
    @Override
    public boolean isDone() {
        if (!super.isDone()) {
            return false;
        }
        Object disruptor = ObjectUtils.mirror().on(eventChannel).get().field("disruptor");
        BasicExecutor executor = (BasicExecutor) ObjectUtils.mirror().on(disruptor).get().field("executor");
        Queue<Thread> threads = (Queue<Thread>) ObjectUtils.mirror().on(executor).get().field("threads");
        for (Thread t : threads) {
            if (!Thread.State.WAITING.equals(t.getState())) {
                return false;
            }
        }
        long cursor = ((Disruptor) disruptor).getCursor();
        Object consumerRepository = ObjectUtils.mirror().on(disruptor).get().field("consumerRepository");
        Object consumerInfos = ObjectUtils.mirror().on(consumerRepository).get().field("consumerInfos");
        Object consumerInfo = ((List) consumerInfos).get(0);
        Object eventprocessor = ObjectUtils.mirror().on(consumerInfo).get().field("eventprocessor");
        Object sequenceBarrier = ObjectUtils.mirror().on(eventprocessor).get().field("sequenceBarrier");
        long sequence = ((SequenceBarrier) sequenceBarrier).getCursor();
        if (cursor != sequence) {
            return false;
        }
        return true;
    }

    /**
     * @throws Exception
     */
    @Override
    public void initialize() throws Exception {
        eventChannel = new DisruptorEventChannel(stepName, inputTopic);
        eventChannel.setEventHandler(eventChannelHandler);
        ((DisruptorEventChannel) eventChannel).setBufferSize(bufferSize);
    }

    /**
     * @throws Exception
     */
    @Override
    public void destroy() throws Exception {

    }
}
